package com.distribute.api;

import com.distribute.ServiceContext;
import com.distribute.config.Config;
import com.distribute.factory.ServiceFactory;
import com.distribute.pojo.Task.*;
import com.distribute.pojo.enums.ConditionType;
import com.distribute.pojo.enums.ReleaseState;
import com.distribute.pojo.enums.ServiceType;
import com.distribute.pojo.enums.TaskType;
import com.distribute.pojo.project.Project;
import com.distribute.pojo.user.Tenant;
import com.distribute.pojo.user.User;
import com.distribute.pojo.workflow.*;
import com.distribute.proxy.DolphinSchedulerAdminProxy;
import com.distribute.utils.JsonUtils;
import config.DSConfig;
import core.DSApiClient;
import core.api.ApiClient;
import org.jetbrains.annotations.NotNull;
import service.DSProjectService;
import service.DSTaskService;
import service.DSWorkflowService;
import service.api.Service;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.*;

public class DolphinSchedulerClient implements ServiceProvider {

    private static volatile DolphinSchedulerClient instance;

    // integrate service
    private final ApiClient apiClient;

    private final Map<ServiceType, Service> serviceMap = new HashMap<>();

    private final Map<TaskType,Class<? extends TaskDefinition>> taskDefinitionMap = new HashMap<>();

    private DolphinSchedulerClient(DSConfig dsConfig) {
        apiClient = new DSApiClient(dsConfig);
        init(apiClient);
    }

    private void init(ApiClient apiClient) {
        ServiceLoader<ServiceFactory> serviceFactories = ServiceLoader.load(ServiceFactory.class);

        // 使用lambda 表达式 将serviceFactories 中的每一个转换成service 并放入servicemap
        serviceFactories.forEach(serviceFactory -> {
            Service service = serviceFactory.createService(apiClient);
            serviceMap.put(service.getServiceType(), service);
        });

        ServiceLoader<TaskDefinition> taskDefinitions = ServiceLoader.load(TaskDefinition.class);
        taskDefinitions.forEach(taskDefinition -> {
            taskDefinitionMap.put(taskDefinition.getTaskType(), taskDefinition.getClass());
        });
    }

    public static DolphinSchedulerClient getInstance(Config config) {
        if (instance == null) {
            synchronized (DolphinSchedulerClient.class) {
                // 从 service context 中获取 admin proxy 分配的用户
                ServiceContext context = ServiceContext.getInstance();
                context.setServiceName(config.getServiceName());

                DolphinSchedulerAdminProxy adminProxy = DolphinSchedulerAdminProxy.getInstance(config);
                Tenant tenant = adminProxy.assignTenant();
                User user = adminProxy.assignUser(tenant);

                context.setUser(user);
                context.setTenant(tenant);

                if (instance == null) {
                    DSConfig dsConfig = new DSConfig(config.getUrl(),
                            user.getUserName(), config.getServiceName() + 123,
                            config.getTimeout(), config.getRetryTimes());
                    instance = new DolphinSchedulerClient(dsConfig);
                }
            }
        }
        return instance;
    }


    @Override
    public Service getService(ServiceType serviceType) {
        Service service = serviceMap.get(serviceType);
        if (service == null) {
            throw new IllegalArgumentException("Unsupported service type: " + serviceType);
        }
        return service;
    }


    @Override
    public WorkflowDefinition generateWorkFlow(WorkflowStruct workFlowStruct) {
        DSProjectService dsProjectService = (DSProjectService) getService(ServiceType.PROJECT);
        Project project = dsProjectService.getProject(workFlowStruct.getProjectName());
        if (project == null) {
            project = dsProjectService.createProject(workFlowStruct.getProjectName(), workFlowStruct.getProjectDescription());
        }
        List<? extends TaskDefinition> taskDefinitions = workFlowStruct.getTaskStructs();
        ArrayList<TaskRelation> relations = new ArrayList<>();
        ArrayList<Location> locationList = new ArrayList<>();
        for (TaskDefinition taskDefinition : taskDefinitions) {
            TaskRelation taskRelation = getTaskRelation(taskDefinition);
            Location location = new Location();
            location.setTaskCode(taskDefinition.getCode());
            location.setX(269);
            location.setY(269);
            relations.add(taskRelation);
            locationList.add(location);
        }
        DSWorkflowService workFlowService = (DSWorkflowService) getService(ServiceType.WORKFLOW);
        workFlowService.verifyWorkflowName(project.getCode(),workFlowStruct.getWorkflowName());
        String taskDefinitionJson = JsonUtils.converString(taskDefinitions);
        String taskRelationJson = JsonUtils.converString(relations);
        String locations = JsonUtils.converString(locationList);
        WorkflowDefinition workFlowDefinition = workFlowService.createWorkflowDefinition(project.getCode(), taskDefinitionJson,
                taskRelationJson, locations, workFlowStruct.getWorkflowName(), ServiceContext.getInstance().getTenant().getTenantCode(),
                workFlowStruct.getWorkflowExecutionType(), workFlowStruct.getWorkflowDescription(), workFlowStruct.getGlobalParams(), workFlowStruct.getTimeout(),
                workFlowStruct.getReleaseState());
        return workFlowDefinition;
    }

    @NotNull
    private static TaskRelation getTaskRelation(TaskDefinition taskDefinition) {
        TaskRelation taskRelation = new TaskRelation();
        if (taskDefinition.getPrev() == null) {
            taskRelation.setPreTaskCode(0L);
            taskRelation.setPreTaskVersion(0);
        } else {
            taskRelation.setPreTaskCode(taskDefinition.getPrev().getCode());
            taskRelation.setPreTaskVersion(taskDefinition.getPrev().getVersion());
        }
        if (taskDefinition.getNext() == null) {
            taskRelation.setPostTaskCode(taskDefinition.getCode());
            taskRelation.setPostTaskVersion(taskDefinition.getVersion());
        } else {
            taskRelation.setPostTaskCode(taskDefinition.getNext().getCode());
            taskRelation.setPostTaskVersion(taskDefinition.getNext().getVersion());
        }
        taskRelation.setConditionType(ConditionType.NONE);
        taskRelation.setConditionParams(null);
        return taskRelation;
    }

    @Override
    public Project createProject(String projectName, String desc) {
        DSProjectService service = (DSProjectService) getService(ServiceType.PROJECT);
        return service.createProject(projectName, desc);
    }

    @Override
    public Project getProject(String projectName) {
        DSProjectService service = (DSProjectService) getService(ServiceType.PROJECT);
        return service.getProject(projectName);
    }

    @Override
    public Boolean onlineWorkflow(WorkflowDefinition workFlowDefinition) {
        DSWorkflowService service = (DSWorkflowService) getService(ServiceType.WORKFLOW);
        return service.updateWorkflowStatus(workFlowDefinition.getProjectCode(), workFlowDefinition.getName(), ReleaseState.ONLINE);
    }

    @Override
    public Boolean offlineWorkflow(WorkflowDefinition workFlowDefinition) {
        DSWorkflowService service = (DSWorkflowService) getService(ServiceType.WORKFLOW);
        return service.updateWorkflowStatus(workFlowDefinition.getProjectCode(), workFlowDefinition.getName(), ReleaseState.OFFLINE);
    }

    @Override
    public TaskDefinition genTaskDefinition(String projectName, TaskType taskType) {
        DSTaskService service = (DSTaskService) getService(ServiceType.TASK);
        Project project = getProject(projectName);
        Long code = service.genTaskCodes(project.getCode(), 1).get(0);
        Class<? extends TaskDefinition> aClass = taskDefinitionMap.get(taskType);
        if(aClass == null){
            throw new IllegalArgumentException("Unsupported task type: " + taskType);
        }
        try {
            Constructor<? extends TaskDefinition> constructor = aClass.getConstructor();
            TaskDefinition taskDefinition = constructor.newInstance();
            taskDefinition.setCode(code);
            return taskDefinition;
        } catch (NoSuchMethodException e) {
            throw new RuntimeException(e);
        } catch (InvocationTargetException e) {
            throw new RuntimeException(e);
        } catch (InstantiationException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public WorkflowScheduler createWorkflowScheduler(ScheduleStruct scheduleStruct) {
        DSWorkflowService service = (DSWorkflowService) getService(ServiceType.WORKFLOW);
        ScheduleStruct.Schedule schedule = scheduleStruct.getSchedule();
        String scheduleString = JsonUtils.converString(schedule);
        WorkflowScheduler workflowScheduler = service.createWorkflowScheduler(scheduleStruct.getProjectCode(), scheduleString,
                scheduleStruct.getFailureStrategy(), scheduleStruct.getWarningType(),
                scheduleStruct.getProcessInstancePriority(), scheduleStruct.getWarningGroupId(), scheduleStruct.getWorkerGroup(),
                scheduleStruct.getEnvironmentCode(), scheduleStruct.getProcessDefinitionCode());
        return workflowScheduler;
    }

    @Override
    public Boolean startWorkflowInstance(WorkflowInstanceStruct workflowInstanceStruct) {
        DSWorkflowService service = (DSWorkflowService) getService(ServiceType.WORKFLOW);
        return service.startWorkflowInstance(
                workflowInstanceStruct.getProjectCode(),
                workflowInstanceStruct.getProcessDefinitionCode(),
                workflowInstanceStruct.getFailureStrategy(),
                workflowInstanceStruct.getWarningType(),
                workflowInstanceStruct.getWarningGroupId(),
                workflowInstanceStruct.getExecType(),
                workflowInstanceStruct.getStartNodeList(),
                workflowInstanceStruct.getTaskDependType(),
                workflowInstanceStruct.getComplementDependentMode(),
                workflowInstanceStruct.getRunMode(), workflowInstanceStruct.getProcessInstancePriority(),
                workflowInstanceStruct.getWorkerGroup(), workflowInstanceStruct.getEnvironmentCode(),
                workflowInstanceStruct.getStartParams(), workflowInstanceStruct.getExpectedParallelismNumber(),
                workflowInstanceStruct.getDryRun(), workflowInstanceStruct.getScheduleTime()
        );
    }


}
